package com.cd;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamStatementSet;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class MysqlCDC2KafkaSQL {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);
        //enable checkpoint
        env.enableCheckpointing(3000);
        // 设置状态的TTL  生产环境设置为最大乱序程度
        //    tableEnvironment.getConfig().setIdleStateRetention(Duration.ofSeconds(905));
        String sourceMysql = "CREATE TABLE user_info_mysql ( " +
                "     user_name STRING, " +
                "     age INT, " +
                "     weight DOUBLE, " +
                "     PRIMARY KEY(user_name) NOT ENFORCED " +
                "     ) WITH ( " +
                "     'connector' = 'mysql-cdc', " +
                "     'hostname' = 'hadoop102', " +
                "     'port' = '3306'," +
                "     'username' = 'root', " +
                "     'password' = '123456', " +
                "     'database-name' = 'data_test', " +
                "     'table-name' = 'user_info')";
        String sinkKafka = "CREATE TABLE user_info_kafka ( " +
                "     user_name STRING, " +
                "     age INT, " +
                "     weight DOUBLE, " +
                "     topic STRING, " +
                "     partition INT, " +
                "     proctime AS PROCTIME() " +
                "     ) WITH ( " +
                "     'connector' = 'upsert-kafka', " +
                "     'topic' = 'user_info_sink', " +
                "     'properties.bootstrap.servers' = 'hadoop102:9092', " +
                "     'format.type' = 'json', " +
                "     'sink.parallelism' = '1' " +
                "     )";
        String sql = "INSERT INTO user_info_kafka SELECT user_name, age, weight, topic, partition FROM user_info_mysql";

        tableEnvironment.executeSql(sourceMysql);
        tableEnvironment.executeSql(sinkKafka);
        tableEnvironment.executeSql(sql);
        env.execute("mysqlCDC2kafka2");



    }
}
